package com.atguigu.actual.edu0417.app.dwd.db;

import com.atguigu.actual.edu0417.app.SqlBaseApp;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: techyang
 * @since: 2023-10-20 10:22
 * @description: DWD层学习域测试事务事实表
 **/
public class DwdStudyExam extends SqlBaseApp {
    public static void main(String[] args) {
        DwdStudyExam dwdStudyExam = new DwdStudyExam();
        dwdStudyExam.base(args);
    }

    @Override
    public void process(StreamTableEnvironment tableEnv) {

        // ⚠️⚠️⚠️ 设置状态的保留时长,若不设置后期会出现状态过大导致内存不够的情况
        // 考虑 业务上的滞后关系 + 传输延迟
        // 测试和测试问题两张表没有业务上的滞后关系,仅考虑传输延迟,10秒钟足矣
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        // 通过kafka连接器读取topic_db中的数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_study_exam_group"));
        // tableEnv.executeSql("select * from topic_db ").print();

        // 过滤出测验问题数据
        // {"database":"edu","table":"test_exam_question","type":"insert","ts":1697633344,"xid":270477,"xoffset":6,"data":{"id":38926,"exam_id":2622,"paper_id":184,"question_id":1768,"user_id":356,"answer":"5752,5753,5754,5755","is_correct":"1","score":7.10,"create_time":"2023-10-18 20:49:04","update_time":null,"deleted":"0"}}
        Table testExamQuestionTable = tableEnv.sqlQuery(" select ts, " +
                " data['id'] id, " +
                " data['exam_id'] exam_id, " +
                " data['paper_id'] paper_id," +
                " data['question_id'] question_id, " +
                " data['user_id'] user_id, " +
                " data['is_correct'] is_correct, " +
                " data['score'] score " +
                " from topic_db " +
                " where `table` = 'test_exam_question' and `type` = 'insert' ");
        tableEnv.createTemporaryView("test_exam_question",testExamQuestionTable);
        // tableEnv.executeSql("select * from test_exam_question ").print();

        // 过滤出测验表
        // {"database":"edu","table":"test_exam","type":"insert","ts":1697706537,"xid":315208,"commit":true,"data":{"id":6346,"paper_id":184,"user_id":754,"score":35.50,"duration_sec":389,"create_time":"2023-10-19 17:08:58","submit_time":"2023-10-19 17:08:58","update_time":null,"deleted":"0"}}
        Table testExamTable = tableEnv.sqlQuery(" select " +
                " data['id'] id, " +
                " data['paper_id'] paper_id, " +
                " data['user_id'] user_id , " +
                " data['score'] score , " +
                " data['duration_sec'] duration_sec " +
                " from topic_db " +
                " where `table` = 'test_exam' and `type` = 'insert' ");
        tableEnv.createTemporaryView("test_exam",testExamTable);
        // tableEnv.executeSql("select * from test_exam ").print();

        // 将test_exam与test_exam_question进行内连接,连接条件为user_id和paper_id
        Table joinTable = tableEnv.sqlQuery(" select " +
                " eq.id, " +
                " eq.exam_id, " +
                " eq.paper_id, " +
                " eq.question_id, " +
                " eq.user_id, " +
                " eq.is_correct, " +
                " eq.score, " +
                " e.score totalScore," +
                " e.duration_sec," +
                " eq.ts  " +
                " from test_exam e ,test_exam_question eq " +
                " where e.user_id = eq.user_id " +
                " and e.paper_id = eq.paper_id ");
        tableEnv.createTemporaryView("join_table",joinTable);
        // tableEnv.executeSql(" select * from join_table ").print();

        // 写入dwd_study_exam主题中
        // 创建动态表
        tableEnv.executeSql(
                " create table dwd_study_exam " +
                        " ( " +
                        "    id string, " +
                        "    exam_id string, " +
                        "    paper_id string, " +
                        "    question_id string,  " +
                        "    user_id string,  " +
                        "    is_correct string,  " +
                        "    score string,  " +
                        "    totalScore string, " +
                        "    duration_sec string," +
                        "    ts string, " +
                        "    primary key (id) not enforced " +
                        " ) " +
                MyKafkaUtil.getUpsertKafkaDDL("dwd_study_exam"));
        // 写入
        tableEnv.executeSql(" insert into dwd_study_exam select * from join_table ");

    }
}
